#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <limits.h>
#include <dirent.h>
#include <sys/time.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/types.h>

#include "data_type.h"
#include "alloc.h"
#include "list.h"
#include "attrlist.h"
#include "memfunc.h"
#include "basefunc.h"
#include "struct_deal.h"
#include "crypto_func.h"
#include "basefunc.h"
#include "memdb.h"
#include "message.h"
#include "dispatch.h"
#include "ex_module.h"
#include "cube_audit.h"

#include "sys_func.h"
#include "message_box.h"
#include "route_tree.h"
#include "router_process_func.h"

int read_dispatch_file(char * file_name,int is_aspect)
{
	int ret;

	int fd;
	int readlen;
	int json_offset;

	int count=0;
	void * root_node;
	void * findlist;
	void * memdb_template ;
	BYTE uuid[DIGEST_SIZE];
	char json_buffer[4097];
	void * policy;
	int finishread=0;
	int leftlen=0;

	fd=open(file_name,O_RDONLY);
	if(fd<0)
		return fd;

	readlen=read(fd,json_buffer,4096);
	if(readlen<0)
		return -EIO;
	json_buffer[readlen]=0;
	if(readlen<4096)
		finishread=1;

	leftlen=readlen;

	json_offset=0;

	while(leftlen>DIGEST_SIZE)
	{
		readlen=json_solve_str(&root_node,json_buffer);
		if(readlen<0)
		{
			print_cubeerr("solve policy %s error!\n",file_name);
			break;
		}
		json_offset+=readlen;
		Memcpy(json_buffer,json_buffer+readlen,leftlen-readlen);
		leftlen-=readlen;
		json_buffer[leftlen]=0;

		if(finishread==0)
		{
			ret=read(fd,json_buffer+leftlen,4096-leftlen);
			if(ret<4096-leftlen)
				finishread=1;
			leftlen+=ret;	
		}	
	
		if(readlen<32)
			continue;

		policy=route_read_policy(root_node);
		if(policy==NULL)
		{
			print_cubeerr("read %d file error!",count);
			break;
		}
		if((route_path_getstate(policy)==POLICY_IGNORE)
			||(route_path_getstate(policy)==POLICY_CLOSE))
		{
			char * policyname =route_path_getname(policy);
			print_cubeaudit("policy %s is ignored!",policyname);
		}
		else
		{
			route_add_policy(policy);
			count++;
		}
	}
	
	//print_cubeaudit("read %d policy succeed!",count);
	close(fd);
	return count;
}

int proc_router_hit_target(void * message,char * local_uuid,char * proc_name)
{
	int ret;
	char conn_uuid[DIGEST_SIZE*2];
	char * receiver = message_get_receiver(message);
	if(receiver ==NULL)
		return -EINVAL;
	if(receiver[0]==':')
		return 1;
	if(receiver[0]=='@')
		return !(strncmp(receiver+1,proc_name,DIGEST_SIZE*2-1));

	ret=comp_proc_uuid(local_uuid,proc_name,conn_uuid);
	return !(strncmp(receiver,conn_uuid,DIGEST_SIZE*2));

}

int proc_audit_route_init()
{
	int ret=1;
	RECORD(CUBE_AUDIT,ROUTE_SELECT) * route_select;
	RECORD(CUBE_AUDIT,ROUTE_FILTER) * route_filter;
	DB_RECORD * db_record;
	
	route_select = memdb_get_first_record(TYPE_PAIR(CUBE_AUDIT,ROUTE_SELECT));
	while(route_select != NULL)
	{
		if(route_select->filter_op == FILTER_SELSEQ)
		{
			route_filter=Talloc0(sizeof(*route_filter));
			if(route_filter ==NULL)
				return -ENOMEM;
			Memcpy(route_filter,route_select,sizeof(*route_select));
			route_filter->seq_counter=route_filter->seq_num;
			db_record = memdb_store(route_filter,TYPE_PAIR(CUBE_AUDIT,ROUTE_FILTER),NULL);
			if(db_record == NULL)
				return -EINVAL;
			//print_bin_data(db_record->head.uuid,32,16);
		}
		route_select = memdb_get_next_record(TYPE_PAIR(CUBE_AUDIT,ROUTE_SELECT));
	}	
	return ret;
}

int proc_audit_init()
{
	int ret;
	char* dir_name = "./message_log";

	ret = mkdir(dir_name,0755);
	if(ret < 0 && errno != EEXIST){
		print_cubeerr("mkdir message_log failed! errno %d\n",errno);
	}else{
		print_cubeaudit("mkdir message_log for ervery router's message store!\n");	
	}

	DIR* dir_stream = opendir(dir_name);
	if(dir_stream == NULL){
		return 0;
	}


	struct dirent *dir_item = NULL;

	while ((dir_item = readdir(dir_stream)) != NULL)
	{
		char path[100] = "./message_log/";
		strcat(path, dir_item->d_name);

		if (strcmp(dir_item->d_name, ".") == 0 || strcmp(dir_item->d_name, "..") == 0)
		{
			continue;
		}

		unlink(path);
	}

	return 0;
}

int proc_audit_route_filter(void * message)
{
	int ret=0;
	RECORD(CUBE_AUDIT,ROUTE_SELECT) * route_select;
	RECORD(CUBE_AUDIT,ROUTE_FILTER) * route_filter;
	DB_RECORD * db_record;
	MSG_HEAD * msg_head;

	msg_head = message_get_head (message);
	if(msg_head == NULL)
		return -EINVAL;
	
	db_record = memdb_get_first(TYPE_PAIR(CUBE_AUDIT,ROUTE_SELECT));
	if(db_record == NULL)
		return 1;
	route_select = db_record->record;
	while(route_select != NULL)
	{
		if(Strncmp(route_select->route_name,msg_head->route,DIGEST_SIZE)==0)
		{
			switch (route_select->filter_op)
			{
				case FILTER_SELALL:
					return 1;
				case FILTER_EXCEPT:
					return 0;
				case FILTER_SELSEQ:
					db_record = memdb_find(db_record->head.uuid,TYPE_PAIR(CUBE_AUDIT,ROUTE_FILTER));
					if(db_record == NULL)
					{
						return 1;
					}
					route_filter = db_record->record;
					if(route_filter == NULL)
						return 1;
					if(Memcmp(route_filter->match_uuid,msg_head->nonce,DIGEST_SIZE)==0)
					{
						if(route_filter->seq_counter==0)
							return 1;
					}	
					else
					{
						if(route_filter->seq_counter>0)
						{
							route_filter->seq_counter--;
							Memcpy(route_filter->match_uuid,msg_head->nonce,DIGEST_SIZE);
							memdb_store_record(db_record);
							if(route_filter->seq_counter==0)
							{
								return 1;
							}
						}
					}
					break;
				default:
					return -EINVAL;	
					
			}
		}
		db_record = memdb_get_next(TYPE_PAIR(CUBE_AUDIT,ROUTE_SELECT));
		if(db_record == NULL)
			return ret;
		route_select = db_record->record;
	}
	
	return ret;
}
int proc_audit_log(void *message)
{
	int ret;
	char *isostr = "\n************************************************************\n";
	char audit_text[4096];
	int fd;

	MSG_HEAD *msg_head = message_get_head(message);
	char filename[64] = "./message_log/";
	strcat(filename, msg_head->route);
	strcat(filename, ".log");

	ret = message_output_json(message, audit_text);
	audit_text[ret] = 0;

	if (access(filename, F_OK) != 0)
	{
		char *beginstr = "\n**************Begin new trust node proc****************************\n";
		fd = open(filename, O_CREAT | O_WRONLY | O_APPEND, 0755);
		if (fd < 0)
			return -ENOENT;
		
		write(fd, beginstr, strlen(beginstr));
		write(fd, audit_text,strlen(audit_text)); 				
		//print_pretty_text(audit_text, fd);
		close(fd);
	}
	else
	{
		fd = open(filename, O_CREAT | O_WRONLY | O_APPEND, 0755);
		if (fd < 0)
			return -ENOENT;
		
		write(fd, isostr, strlen(isostr)); 				
		write(fd, audit_text,strlen(audit_text)); 				
		//print_pretty_text(audit_text, fd);
		close(fd);
 	}
}
			
int proc_router_recv_msg(void * message,char * local_uuid,char * proc_name)
{
	void * sec_sub;
	int ret;
	MSG_HEAD * msg_head;

	if(message_get_state(message) & MSG_FLAG_LOCAL)
		return 0;

	if(message_get_state(message) & MSG_FLOW_DELIVER)
	{
		return 0;
	}

	if(message_get_state(message) & MSG_FLOW_RESPONSE)
	{
		
//		ret=route_check_sitestack(message);
		if(ret<0)
			return ret;
		// if response stack finished, set the state to FINISH 
		if(ret==0)
			message_set_state(message, MSG_FLOW_FINISH);
	}
	if(message_get_flow(message) & MSG_FLOW_ASPECT)
	{
		
//		ret=route_check_aspectstack(message);
		if(ret<0)
			return ret;
		if(ret==0)
		{
			// if aspect stack finished, remove the aspect flag from state 
			int flow=message_get_flow(message) &(~MSG_FLOW_ASPECT);
			message_set_flow(message, flow);
		}
	}
	return 0;
}

int proc_router_send_msg(void * message,char * local_uuid,char * proc_name)
{
	void * sec_sub;
	int ret;
	MSG_HEAD * msg_head;
	BYTE conn_uuid[DIGEST_SIZE*2];
	if(message_get_flag(message) & MSG_FLAG_LOCAL)
	{
		
			msg_head=message_get_head(message);
			if(msg_head==NULL)
			{
				return  -EINVAL;
			}
			ret=find_ex_module(msg_head->receiver_uuid,&sec_sub);	
			if(sec_sub!=NULL)
			{
//				if(sec_subject_getprocstate(sec_sub)<SEC_PROC_START)
//				{	
//					print_cubeaudit("start process %s!\n",sec_subject_getname(sec_sub));
  //  					ret=sec_subject_start(sec_sub,NULL);
//				}
				send_ex_module_msg(sec_sub,message);
			}
			else
			{
				print_cubeerr("proc_router_send_msg: can't find local message target %s",msg_head->receiver_uuid);
			}
	}
	else
	{
	
			ret=find_ex_module("connector_proc",&sec_sub);	
			if(sec_sub==NULL)
			{
				print_cubeerr("can't find conn process!\n");
				return -EINVAL;
			}
			send_ex_module_msg(sec_sub,message);
			print_cubeaudit("send message to conn process!");
				
	}
	return 0;
}

int proc_router_init(void * sub_proc,void * para)
{

    int ret;
    // main proc: read router config	
    char * config_filename= "./dispatch_policy.json";
    char * aspect_filename=NULL;
   if(para!=NULL)
   {
	struct router_init_para * init_para=para;
        config_filename= init_para->router_file;
        aspect_filename= init_para->aspect_file;
    }	

 // router_policy_init();
    ret=read_dispatch_file(config_filename,0);	
    if(ret<=0)
    {
	    print_cubeerr("read router policy error %d!",ret);
//	    return ret;
    }
	else
		print_cubeaudit("read %d policy from router_policy file!",ret);
    ret=read_dispatch_file(aspect_filename,1);	
    if(ret<=0)
    {
	    print_cubewarn("read aspect policy failed %d!",ret);
//	    return ret;
    }
	else
		print_cubeaudit("read %d policy from aspect policy file!",ret);

    proc_audit_init();
    proc_audit_route_init();
    return 0;
}


int proc_router_start(void * sub_proc,void * para)
{
	int ret;
	int retval;
	struct message_box * message_box;
	MSG_HEAD * message_head;
	void * context;
	int i,j;
	char local_uuid[DIGEST_SIZE*2+1];
	char proc_name[DIGEST_SIZE*2+1];
	char receiver_uuid[DIGEST_SIZE*2+1];
	BYTE conn_uuid[DIGEST_SIZE*2];
	
	ret=proc_share_data_getvalue("uuid",local_uuid);
	ret=proc_share_data_getvalue("proc_name",proc_name);



//	struct timeval time_val={0,10*1000};
	struct timeval router_val;
	router_val.tv_sec=time_val.tv_sec;
	router_val.tv_usec=time_val.tv_usec;
	comp_proc_uuid(local_uuid,proc_name,conn_uuid);

	// message routing loop
	

	while(1)
	{
	// throughout all the sub_proc
		
		usleep(router_val.tv_usec);

		void * sub_proc;
		char * receiver_proc;
		char * sender_proc;
		void * msg_policy;
		void * aspect_policy;
		char aspect_proc[DIGEST_SIZE*2];
		char  origin_proc[DIGEST_SIZE];
		struct message_box * message;
		MSG_HEAD * msg_head;

		if(proc_share_data_getstate()<PROC_LOCAL_START)
			continue;
		msg_policy=NULL;
		ret=get_first_ex_module(&sub_proc);
		while(sub_proc!=NULL)
		{
			void * router_rule;
			int state;
			int flow;

			int send_state;
			int hit_target;

			enum module_type curr_proc_type;

			// this pro is a port proc
			curr_proc_type=ex_module_gettype(sub_proc);

			// enum module,check module and get module's sending message 
			ret=recv_ex_module_msg(sub_proc,&message);
			if(ret<0)
			{
				// get next sub_proc
				get_next_ex_module(&sub_proc);
				continue;
			}
			else if((message==NULL) ||IS_ERR(message))
			{
				get_next_ex_module(&sub_proc);
				continue;	
			}

				// get message's sender as origin_proc
			Strncpy(origin_proc,ex_module_getname(sub_proc),DIGEST_SIZE);
			print_cubeaudit("router get proc %.64s's message ",origin_proc); 

				// if origin message is not		message's value reset
			if(Strcmp(origin_proc,"connector_proc")!=0)
			{
				message_set_sender(message,origin_proc);
			}
			else
			{
				msg_head=message_get_head(message);
				if(Memcmp(msg_head->sender_uuid,EMPTY_UUID,DIGEST_SIZE)==0)
				{
					message_set_sender(message,origin_proc);
				}
			}
		
			// if message is init message

			router_dup_activemsg_info(message);

			message_set_activemsg(message,message);
			msg_head=message_get_head(message);
			// set rjump's value
			switch(ex_module_gettype(sub_proc))
			{
				case MOD_TYPE_CONN:
					msg_head->ljump=1;	
					if(msg_head->state!=MSG_STATE_RESPONSE)
						msg_head->rjump++;
					else
						msg_head->rjump--;
					break;
				case MOD_TYPE_PORT:
					msg_head->ljump=1;	
					msg_head->rjump=1;
					break;
				default:
					msg_head->ljump++;	
					if(msg_head->rjump==0)
						msg_head->rjump=1;
					break;
			}
		
			// enter the message route match process
			
			if(message->policy!=NULL)
			// if message has path already
			{

				_waiting_message_add(message);
			}
			// if message is init message
			else if(msg_head->flow == 0)
			{
				// finding the match policy 
				ret=router_find_route_policy(message,&msg_policy);
				if(ret<0)
				{
					print_cubeerr("Fatal error in finding policy !");
					return ret;
				}
				// if can't find match policy
				if(msg_policy==NULL)
				{
					msg_head->flow=MSG_FLOW_FINISH;
					proc_audit_log(message);
					get_next_ex_module(&sub_proc);
					continue;
				}
				// show the debug info
				print_cubeaudit("new msg match path %.64s's policy",route_path_getname(msg_policy)); 
				message_route_setstart(message,msg_policy);
				_waiting_message_add(message);
			}	
			else if ((msg_head->state & MSG_STATE_RESPONSE)==0)
			// message is not return message 
			{
				ret=message_route_setremotestart(message);
				if(ret<0)
				{
					print_cubeerr("Fatal error in remotestart!");
					return ret;
				}
				if(message->policy==NULL)
				{
					msg_head->flow=MSG_FLOW_FINISH;
					proc_audit_log(message);
					get_next_ex_module(&sub_proc);
					continue;
				}
				print_cubeaudit("remote msg match path %.64s's policy",msg_head->route);
				_waiting_message_add(message);
			}
			// message is a return message
			else
			{
				// we should find message's response record and reload the path
				TRACE_NODE * trace_node;
				trace_node=message_route_findtrace(message,MSG_FLOW_QUERY);
				if(trace_node == NULL)
				{
					proc_audit_log(message);
					print_cubeerr("Fatal error in remotestart!");
					usleep(1000);
					return ret;
				}
				ret= message_route_settracetarget(message,trace_node);
				if(message->policy==NULL)
				{
				    print_cubeaudit("maybe an aspect remote msg %.64s come",msg_head->route);
					proc_audit_log(message);
				}
                else
                {

				    print_cubeaudit("response remote msg match path %.64s's policy",msg_head->route);
                }
				print_cubeaudit("router_proc: add message to send queue");
				_waiting_message_add(message);

			} 
			get_next_ex_module(&sub_proc);
		};
		
		message=_waiting_message_removehead();

		int issend;
		int isaspect;
		while(message!=NULL)
		{		
			issend=0;
			isaspect=0;

			msg_head=message_get_head(message);
			if(msg_head->flow==MSG_FLOW_DELIVER)
			{
				ret=message_route_setnext(message);
				if(ret!=0)
				{
					// there is a target in the next	
					proc_audit_log(message);
					issend=1;
				}	
				else
				{
					message->head.flow=MSG_FLOW_FINISH;
					proc_audit_log(message);
				}
			}
			else if(msg_head->flow == MSG_FLOW_QUERY)
			{
				ret=message_route_setnext(message);
				if(ret < 0)
				{
					print_cubeerr("route query error!\n");
				}	
				else if((ret==0) ||(ret == ROUTE_TARGET_FINISH))
				{
					if(msg_head->rjump>1)
					{
						TRACE_NODE * trace_node;
						trace_node=message_route_findtrace(message,MSG_FLOW_QUERY);
						if(trace_node == NULL)
						{
							proc_audit_log(message);
							print_cubeerr("find trace message (%d %d)failed!\n",message->head.record_type,message->head.record_subtype);
						}
						else
						{
							if((msg_head->state & MSG_STATE_RESPONSE)||(ret==ROUTE_TARGET_FINISH))  
							{
								ret= message_route_settracesrc(message,trace_node);
							}
							else
							{
								ret= message_route_gettracetarget(message,trace_node);
							}
							proc_audit_log(message);
							issend=1;
						}
					}
					else
					{
						ASPECT_NODE * aspect_node;
						aspect_node=message_route_findtrace(message,MSG_FLOW_ASPECT);
						if(aspect_node == NULL)   // find aspect error!
						{
							proc_audit_log(message);
							print_cubeerr("find aspect  message (%d %d)failed!\n",message->head.record_type,message->head.record_subtype);
						}
                        else              // find an aspect node  
                        {
							ret= message_route_aspectrecover(message,aspect_node);
							proc_audit_log(message);
							issend=1;
							isaspect=1;
                        }
					}
				}
				else
				{
					// this is not the end	
					proc_audit_log(message);
					issend=1;
				}
				
			}
			else
			{
				print_cubeerr("message route is nor deliver or query mode");
				return -EINVAL;
			}
			if(issend==1)
			{
				issend=0;
				
				ROUTE_NODE * curr_pathnode;
				BRANCH_NODE * curr_branch;
				Record_List * curr_record;

				curr_record=message->path_site;

				if((curr_record !=NULL) && (curr_record->record!=NULL))
				{
					curr_pathnode=curr_record->record;
					curr_record = _node_list_getfirst(&curr_pathnode->aspect_branch);
	
					while((curr_record!=NULL) &&(curr_record->record!=NULL))
					{
						curr_branch = curr_record->record;
						switch(curr_branch->branch_type)
						{
							case MSG_FLOW_DUP:
								ret=_route_match_aspect_branch(message,curr_branch);	
								if(ret>0)
								{
									struct message_box * new_msg = route_dup_message(message,curr_branch);	
									print_cubeaudit("dup a message to path %.64s's policy",new_msg->head.route);
									_waiting_message_add(new_msg);
								}
								break;
							case MSG_FLOW_ASPECT:
								if(isaspect)
									break;
								ret=_route_match_aspect_branch(message,curr_branch);	
								if(ret>0)
								{
									ret=route_aspect_message(message,curr_branch);
									print_cubeaudit("aspect message to path %.64s's policy",message->head.route);
									_waiting_message_add(message);
									message=NULL;
								}
								break;
							default:
								print_cubeaudit("message branch is nor dup no aspect");
								break;
						}
						curr_record = _node_list_getnext(&curr_pathnode->aspect_branch);
					}	
				}	
				if(message!=NULL)
				{
					if(Memcmp(message->head.receiver_uuid,EMPTY_UUID,DIGEST_SIZE)!=0)
					{
						message->active_msg = message;
						ret=proc_router_send_msg(message,local_uuid,proc_name);
					}
				}
			}
			message=_waiting_message_removehead();
		}

			// test's end
	}
	return 0;
}
